package com.chengv.redismq.core;

import cn.hutool.core.lang.Assert;
import com.chengv.redismq.constant.RedisMQConstant;
import com.chengv.redismq.pojo.MQMessage;
import lombok.Getter;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;

import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * Redis MQ 模板
 *
 * @author chengv
 */
public class RedisMQTemplate {

    @Getter
    private final RedisTemplate<String, String> redisTemplate;
    @Getter
    private final RedisSerializer<Object> hashValueSerializer;

    public RedisMQTemplate(RedisTemplate<String, String> redisTemplate, RedisSerializer<Object> hashValueSerializer) {
        this.redisTemplate = redisTemplate;
        this.hashValueSerializer = hashValueSerializer;
    }

    /**
     * 发送消息
     *
     * @param topic   消息主题
     * @param payload 消息负载
     */
    public void send(String topic, Object payload) {
        Assert.notBlank(topic, "topic 不能为空");

        ObjectRecord<String, String> record = StreamRecords.newRecord()
                .ofObject(convertMQMessageAndSerialize(payload)).withStreamKey(RedisMQConstant.QUEUE_PREFIX + topic);
        redisTemplate.opsForStream().add(record);
    }

    /**
     * 发送延时消息
     *
     * @param topic    消息主题
     * @param payload  消息负载
     * @param delay    延迟时间
     * @param timeUnit 延迟时间单位
     */
    public void send(String topic, Object payload, long delay, TimeUnit timeUnit) {
        Assert.notBlank(topic, "topic 不能为空");
        Assert.isTrue(delay > 0, "delay 必须大于0");
        Assert.notNull(timeUnit, "timeUnit 不能为空");

        redisTemplate.opsForZSet().add(RedisMQConstant.DELAY_QUEUE_PREFIX + topic, convertMQMessageAndSerialize(payload),
                System.currentTimeMillis() + timeUnit.toMillis(delay));
    }

    /**
     * 将消息负载转换为 MQ 消息实体，原始消息会被放到 {@link MQMessage} 的 payload 字段中，然后序列化为 JSON 字符串
     *
     * @param payload 消息负载
     * @return MQ 消息实体 JSON 字符串
     */
    private String convertMQMessageAndSerialize(Object payload) {
        MQMessage mqMessage = new MQMessage(payload);
        return new String(Objects.requireNonNull(hashValueSerializer.serialize(mqMessage)));
    }
}